package com.es.test.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch._types.query_dsl.*;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.PointInTimeReference;
import co.elastic.clients.elasticsearch.indices.RefreshRequest;
import co.elastic.clients.json.JsonData;
import com.alibaba.excel.util.StringUtils;
import com.alibaba.fastjson2.JSON;
import com.es.test.contants.IndexName;
import com.es.test.document.LabelDocument;
import com.es.test.dto.LabelDto;
import com.es.test.utils.LabelBuilder;
import com.es.test.utils.ReadExcelData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.io.IOException;
import java.text.ParseException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;


/**
 * @author django
 * @date 2024/2/1
 * @description TODO
 */
@Slf4j
@Service
public class UserLabelService {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
            100,
            100,
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10)
    );

    public AtomicLong atomicLong = new AtomicLong(0);
    public AtomicLong timeLong = new AtomicLong(0);
    public AtomicLong nLong = new AtomicLong(0);
    public AtomicLong wLong = new AtomicLong(0);

    @Autowired
    private ElasticsearchClient elasticsearchClient;


    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserAsyncService userAsyncService;

    public void redisBitMap() {
        // 随机产生2亿条数据，然后去交集

        BitSet bitSet1 = new BitSet();
        BitSet bitSet2 = new BitSet();
        Random random = new Random();


        redisTemplate.execute(new SessionCallback<List<Object>>() {
            @Override
            public List<Object> execute(RedisOperations operations) throws DataAccessException {
                // 开始事务
                operations.multi();

                // 命令集合
              /*  for (int i = 0; i < 300; i++) {
                    int randomIndex = random.nextInt(300);
                    redisTemplate.opsForValue().setBit("bitSet1", randomIndex, true);
                    int randomIndex2 = random.nextInt(300);
                    redisTemplate.opsForValue().setBit("bitSet2", randomIndex2, true);
                    System.out.println("bitSet1----" + randomIndex + "---bitSet2----" +  randomIndex2);
                }*/
                redisTemplate.opsForValue().setBit("bitSet1", 2, true);
                redisTemplate.opsForValue().setBit("bitSet1", 4, true);
                redisTemplate.opsForValue().setBit("bitSet2", 4, true);
                redisTemplate.opsForValue().setBit("bitSet2", 2, true);
                return operations.exec();
            }
        });
        bitSet1.set(2);
        bitSet1.set(4);
        bitSet2.set(2);
        bitSet2.set(4);
        redisTemplate.opsForValue().set("bitSet5", new String(bitSet1.toByteArray()));

        redisTemplate.opsForValue().set("bitSet6", new String(bitSet1.toByteArray()));

        // 求合集
        RedisConnection connection1 = redisTemplate.getConnectionFactory().getConnection();
        Long aLong = connection1.bitOp(RedisStringCommands.BitOperation.AND,
                "bitSet3".getBytes(),
                "bitSet1".getBytes(),
                "bitSet2".getBytes());


        Long aLong1 = connection1.bitOp(RedisStringCommands.BitOperation.AND,
                "bitSet7".getBytes(),
                "bitSet5".getBytes(),
                "bitSet6".getBytes());

        String bitSet3 = redisTemplate.opsForValue().get("bitSet3");

        BitSet bitSet = BitSet.valueOf(bitSet3.getBytes());
        for (int i = bitSet.nextSetBit(0); i >= 0; i = bitSet.nextSetBit(i + 1)) {
            System.out.print(i - 1 + " ");
        }


        String bitSet7 = redisTemplate.opsForValue().get("bitSet7");

        BitSet bitMap7 = BitSet.valueOf(bitSet7.getBytes());
        for (int i = bitMap7.nextSetBit(0); i >= 0; i = bitMap7.nextSetBit(i + 1)) {
            System.out.print(i + " ");
        }
         /*
        BitSet bit3 = new BitSet(bitSet3.getBytes().length);
        String sssss = convertStringToBinary(bitSet3);
        BitSet bitSet = BitSet.valueOf(sssss.getBytes());
        /* String s = convertStringToBinary(bitSet3);
        BitSet bitSet = BitSet.valueOf(s.getBytes());
        bitSet.stream().forEach(item ->{
            System.out.println(item);
        });
        bit3.stream().forEach(item -> {
            if (item > 0) {
                System.out.println("命中----》" + item);
            }
        });
        for (int i = bitSet.nextSetBit(0); i >= 0; i = bitSet.nextSetBit(i + 1)) {
            System.out.print(i + " ");
        }
        char[] chars = sssss.toCharArray();
        for (int i = 0; i < chars.length; i++) {
            if ((int) chars[i] == '1') {
                System.out.println("--------------" + i);
            }
        }*/

    }

    public static void main(String[] args) {
        byte[] bytes = "H".getBytes();
        String cat = convertStringToBinary("H");
        char[] chars = cat.toCharArray();
        for (int i = 0; i < chars.length; i++) {
            char aChar = chars[i];
            System.out.println(aChar == '1');
        }
        BitSet bitSet = BitSet.valueOf("H".getBytes());
        bitSet.stream().forEach(System.out::println);
        System.out.println(bitSet);

        BitSet bs = new BitSet();
        bs.set(1);
        bs.set(4);
        System.out.println(bs);
    }

    public static String convertStringToBinary(String inputString) {
        StringBuilder binaryStringBuilder = new StringBuilder();

        for (char character : inputString.toCharArray()) {
            String binaryChar = Integer.toBinaryString(character);

            // Ensure each binary representation is 8 bits long
            while (binaryChar.length() < 8) {
                binaryChar = "0" + binaryChar;
            }

            binaryStringBuilder.append(binaryChar);
        }

        return binaryStringBuilder.toString();
    }

    private void generateAndSetBitSet(String bitSetKey, int totalDataPoints, int positiveSamples) {
        redisTemplate.execute(new SessionCallback<List<Object>>() {
            @Override
            public List<Object> execute(RedisOperations operations) throws DataAccessException {
                // Start a transaction
                operations.multi();

                // Set the positive samples in the BitSet
                for (int i = 0; i < positiveSamples; i++) {
                    int randomIndex = i;  // In a real-world scenario, generate random indices
                    operations.opsForValue().setBit(bitSetKey, randomIndex, true);
                }

                // Execute the transaction
                return operations.exec();
            }
        });
    }


    public void operate(int type) {
        BitSet bitSet1 = new BitSet();
        BitSet bitSet2 = new BitSet();
        LabelDto label1 = null;
        LabelDto label2 = null;
        if (type == 1) {
            System.out.println("choose type 1");
            label1 = new LabelDto("c_school_name_last", "北京工商管理专修学院");
            label2 = new LabelDto("last_login_cooperator", "自然流量");
        } else if (type == 2) {
            System.out.println("choose type 2");
            label1 = new LabelDto("yjs_major", "西南科技大学");
            label2 = new LabelDto("candidatepopular", "推送次数≤10次");
        } else if (type == 3) {
            System.out.println("choose type 3");
            label1 = new LabelDto("c_school_name_last", "北京工商管理专修学院");
            label2 = new LabelDto("last_login_cooperator", "自然流量");
        } else {
            System.out.println("choose type other");
            label1 = new LabelDto("yjs_degree", "优秀大学");
            label2 = new LabelDto("resume_key_word", "销售总监");
        }

        long startTime = System.currentTimeMillis();
        try {
            performScrollQuery(IndexName.USER_LABEL, label1, bitSet1, startTime);
            performScrollQuery(IndexName.USER_LABEL, label2, bitSet2, startTime);
            atomicLong.set(0);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        bitSet1.and(bitSet2);
        bitSet1 = null;
        bitSet2 = null;
        System.out.println(bitSet1);
        AtomicInteger n = new AtomicInteger();
        bitSet1.stream().parallel().forEach(item -> {
            if (item > 0) {
                n.getAndIncrement();
            }
        });
        System.out.println("operate result" + n.get());

    }

    /* public String generateData(int start, int end) {
         System.out.println("start batch insert.............");
         // 手动创建线程池
         ExecutorService customThreadPool = Executors.newFixedThreadPool(4);


         Map<String, List<String>> excelMap = ReadExcelData.getExcelMap();
         List<CompletableFuture> futures = new ArrayList<>();
         int n = 0;
            //         for (int i = 11022; i < 20000; i++) {
          for (int i = start; i < end; i++) {
             System.out.println(LocalDateTime.now() + "i--------------------------------------->" + i);

             int finalI = i;
              try {
                 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                     return handle(finalI, excelMap);
                 }, customThreadPool);
                  futures.add(future);
                 n ++;
                  if (futures.size() > 5) {
                      for (CompletableFuture completableFuture : futures) {
                          completableFuture.get();
                      }
                      futures.clear();
                  }
             } catch (Exception e) {
                  // TODO这里一定捕获异常并睡眠一会
             }
         }

         return "success";
     }*/
    public String generateData(int start, int end, int threadNum) {
        System.out.println("start batch insert.............");
        // 手动创建线程池
        ExecutorService customThreadPool = Executors.newFixedThreadPool(threadNum);


        long startTime = System.currentTimeMillis();
        Map<String, List<String>> excelMap = ReadExcelData.getExcelMap();
        int n = 0;
        for (int i = start; i < end; i++) {
            int finalI = i;
            try {
                CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                    long subStartTime = System.currentTimeMillis();
                    return handle(finalI, excelMap, startTime, subStartTime);
                }, customThreadPool);

            } catch (Exception e) {
                log.info("error------------------->", e);
                // TODO这里一定捕获异常并睡眠一会
            }
        }

        return "success";
    }


    private String handle(int i, Map<String, List<String>> excelMap, long startTime, long subStartTime) {


        BulkRequest.Builder br = new BulkRequest.Builder();

        for (int j = 0; j < 10000; j++) {
            LabelDocument labelDocument = new LabelDocument();
            labelDocument.setId(10000 * i + j);
            labelDocument.setTimestamp(System.currentTimeMillis());
            labelDocument.setName("name-".concat(String.valueOf(i)));
            Map map = JSON.parseObject(JSON.toJSONString(labelDocument), Map.class);


            if (j % 100 == 1) {// 2 0000 0000 /100 = 20000 00
                map.put("test_label_1", "最近1天登录");
                map.put("test_label_2", "最近2天登录");
                map.put("test_label_3", "最近3天登录");
                map.put("test_label_4", "最近4天登录");
                map.put("test_label_5", "最近5天登录");
                map.put("test_label_6", "最近6天登录");
                map.put("test_label_7", "最近7天登录");
                map.put("test_label_8", "最近8天登录");
                map.put("test_label_9", "最近9天登录");
                map.put("test_label_10", "最近10天登录");
                map.put("test_label_11", "最近11天登录");
                map.put("test_label_12", "最近12天登录");
                map.put("test_label_13", "最近13天登录");
                map.put("test_label_14", "最近14天登录");
                map.put("test_label_15", "最近15天登录");
                map.put("test_label_16", "最近16天登录");
                map.put("test_label_17", "最近17天登录");
                map.put("test_label_18", "最近18天登录");
                map.put("test_label_19", "最近19天登录");
                map.put("test_label_20", "最近20天登录");
                map.put("test_label_num_c", j + i - j / 2);
            }
            if (j % 1000 == 1) {// 2 0000 0000 /100 = 20000 0

                map.put("test_1", "工作1年");
                map.put("test_2", "工作2年");
                map.put("test_3", "工作3年");
                map.put("test_4", "工作4年");
                map.put("test_5", "工作5年");
                map.put("test_6", "工作6年");
                map.put("test_7", "工作7年");
                map.put("test_8", "工作8年");
                map.put("test_9", "工作9年");
                map.put("test_10", "工作10年");
                map.put("test_label_num_d", j / 2 + j);
                map.put("test_label_num_a", 1);
            }
            if (j % 888 == 1) {
                map.put("test_label_num_a", 2);
            }

            Random random = new Random();
            // 生成10以内的随机整数
            int randomNumber = random.nextInt(72);
            List<String> labels = LabelBuilder.getRandomElements(LabelBuilder.labelList, randomNumber);
            labels.forEach(item -> {
                List<String> labelVueList = excelMap.get(item);
                try {
                    itemHandle(map, item, labelVueList);
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            });

            br.operations(op -> op
                    .index(idx -> idx
                            .index(IndexName.USER_LABEL)
                            .id(String.valueOf(labelDocument.getId()))
                            .document(map)
                    ));

        }
        try {
            System.out.println(Thread.currentThread().getName() + "--start---->" + i);
            long endTime = System.currentTimeMillis();
           /* if ((endTime - startTime) / 1000 >= 3600) {
                Thread.sleep(2000000000000000L);
            }*/


            elasticsearchClient.bulk(br.build());
            wLong.addAndGet(10000);
            log.info("bulk insert avg--->{}-{}", wLong.get() / ((System.currentTimeMillis() - startTime) / 1000), wLong.get());
//            userAsyncService.bulk(br.build());
            List<String> list = new ArrayList<>();
            list.add(IndexName.USER_LABEL);
            RefreshRequest.Builder index = new RefreshRequest.Builder().index(list);
//            RefreshRequest refreshRequest = index.build();
//            elasticsearchClient.indices().refresh(refreshRequest);
            System.out.println(Thread.currentThread().getName() + "-finished---->" + i);
        } catch (Exception e) {
            log.error("---------------", e);
        }
        //elasticsearchTemplate.bulk(queries, IndexCoordinates.of(IndexName.USER_LABEL))
        //elasticsearchTemplate.bulkIndex(queries, IndexCoordinates.of(IndexName.USER_LABEL));
        // elasticsearchTemplate.indexOps( IndexCoordinates.of(IndexName.USER_LABEL)).refresh();

        return String.valueOf(i) + "-finish";
    }


    private void itemHandle(Map map, String item, List<String> labelVueList) throws ParseException {
        if (CollectionUtils.isNotEmpty(labelVueList)) {
            List<String> randomElements = LabelBuilder.getRandomElements(labelVueList, 1);
            String data = randomElements.get(0);
            if (data.contains(".") && data.split("-").length == 3 && data.split(":").length == 3) {
                map.put(item, DateUtils.parseDateStrictly(data, "yyyy-MM-dd HH:mm:ss.s").getTime());

            } else if (data.split("-").length == 3) {
                map.put(item, DateUtils.parseDateStrictly(data, "yyyy-MM-dd HH:mm:ss").getTime());
            } else {
                map.put(item, data);
            }

        }
    }

    /**
     * 两个标签取并集
     */


    private void performScrollQuery(String indexName, LabelDto label, BitSet bitSet, long startTime) throws IOException {


        CountResponse count = elasticsearchClient.count(s -> s.index(indexName).query(t -> t.term(v -> v.field(label.getName()).value(label.getValue()))));

        log.info("start-total-num----" + count.count() + "-------------" + JSON.toJSONString(label));
        long total = atomicLong.addAndGet(count.count());
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            SearchResponse<HashMap> searchResponse = elasticsearchClient.search(s -> s
                            .index(IndexName.USER_LABEL)
                            .query(q -> q
                                    .term(t -> t
                                            .field(label.getName())
                                            .value(label.getValue())
                                    )
                            ).trackScores(false).source(sc -> sc.fetch(false)).size(10000).scroll(t -> t.time("100s")),
                    HashMap.class
            );
            timeLong.addAndGet(searchResponse.hits().hits().size());

            //使用scroll深度分页查询

            // 处理第一次查询结果
            processSearchHitsListHit(searchResponse.hits().hits(), bitSet);
            // 获取滚动 ID
            String scrollId = searchResponse.scrollId();
            long beginTime = System.currentTimeMillis();
            ScrollResponse<HashMap> scrollSearch = null;
            int n = 1;

            do {
                String finalScrollId = scrollId;
                long st = System.currentTimeMillis();
                try {
                    scrollSearch = elasticsearchClient.scroll(s -> s.scrollId(finalScrollId).scroll(t -> t.time("100s")), HashMap.class);

                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ElasticsearchException e) {
                    log.error("query one start--->", e);
                }
                n++;
                nLong.addAndGet(1);
                processSearchHitsListHit(scrollSearch.hits().hits(), bitSet);
                timeLong.addAndGet(scrollSearch.hits().hits().size());

                log.info(" {}----start----{}---->cost-{}--query-one--time-avg--{}", Thread.currentThread().getName(),
                        n, (st - System.currentTimeMillis()), (startTime - System.currentTimeMillis()) / nLong.get());
                log.info("avg=========>total------->{}->,avg------>{}/s ", timeLong.get(), (timeLong.get() / ((System.currentTimeMillis() - startTime) / 1000)));

                scrollId = scrollSearch.scrollId();
            } while (!scrollSearch.hits().hits().isEmpty());

            // 清除滚动上下文
            //清除 scroll
            String finalScrollId1 = scrollId;
            elasticsearchClient.clearScroll(c -> c.scrollId(finalScrollId1));


        } catch (Exception e) {
            e.printStackTrace();
        }
        stopWatch.stop();
        log.info("total time count(s):---->" + stopWatch.getTotalTimeSeconds());
    }

    private void complexQuery(String indexName, Query query, BitSet bitSet, long startTime) throws IOException {


        CountResponse count = elasticsearchClient.count(s -> s.index(indexName).query(query));

        log.info("start-total-num----" + count.count());
        long total = atomicLong.addAndGet(count.count());
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            SearchResponse<HashMap> searchResponse = elasticsearchClient.search(s -> s
                            .index(IndexName.USER_LABEL)
                            .query(query).trackScores(false).source(sc -> sc.fetch(false)).size(10000).scroll(t -> t.time("10s")),
                    HashMap.class
            );
            timeLong.addAndGet(searchResponse.hits().hits().size());

            //使用scroll深度分页查询

            // 处理第一次查询结果
            //processSearchHitsListHit(searchResponse.hits().hits(), bitSet);
            // 获取滚动 ID
            String scrollId = searchResponse.scrollId();
            long beginTime = System.currentTimeMillis();
            ScrollResponse<HashMap> scrollSearch = null;
            int n = 1;

            do {
                String finalScrollId = scrollId;
                long st = System.currentTimeMillis();
                try {
                    scrollSearch = elasticsearchClient.scroll(s -> s.scrollId(finalScrollId).scroll(t -> t.time("100s")), HashMap.class);

                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ElasticsearchException e) {
                    log.error("query one start--->", e);
                }
                if (scrollSearch.hits().hits().size() == 0) {
                    return;
                }
                n++;
                nLong.addAndGet(1);
                //processSearchHitsListHit(scrollSearch.hits().hits(), bitSet);
                timeLong.addAndGet(scrollSearch.hits().hits().size());

                log.info(" {}----start----{}---->cost#############{}--query-one--time-avg--{}", Thread.currentThread().getName(),
                        n, (st - System.currentTimeMillis()), (startTime - System.currentTimeMillis()) / nLong.get());
                log.info("avg=========>total------->{}->,avg------>{}/s ", timeLong.get(), (timeLong.get() / ((System.currentTimeMillis() - startTime) / 1000)));

                scrollId = scrollSearch.scrollId();
            } while (!scrollSearch.hits().hits().isEmpty());

            // 清除滚动上下文
            //清除 scroll
            String finalScrollId1 = scrollId;
            elasticsearchClient.clearScroll(c -> c.scrollId(finalScrollId1));


        } catch (Exception e) {
            e.printStackTrace();
        }
        stopWatch.stop();
        log.info("total time count(s):---->" + stopWatch.getTotalTimeSeconds());
    }

    Set<String> set = new HashSet<>();

    private void pitComplexQuery(String indexName, Query query, BitSet bitSet, long startTime) throws IOException {


        CountResponse count = elasticsearchClient.count(s -> s.index(indexName).query(query));

        log.info("start-total-num----" + count.count());
        long total = atomicLong.addAndGet(count.count());
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();


        try {
            OpenPointInTimeRequest.Builder openBuilder = new OpenPointInTimeRequest.Builder();
            openBuilder.index(IndexName.USER_LABEL);
            openBuilder.keepAlive(t -> t.time("30m"));
            String id = elasticsearchClient.openPointInTime(openBuilder.build()).id();


            //使用scroll深度分页查询

            // 处理第一次查询结果
            //processSearchHitsListHit(searchResponse.hits().hits(), bitSet);
            // 获取滚动 ID
            long beginTime = System.currentTimeMillis();
            SearchResponse<HashMap> pidSearch = null;
            int n = 1;
            List<FieldValue> sort = new ArrayList<>();
            do {
                long st = System.currentTimeMillis();
                try {

                    PointInTimeReference.Builder pBuilder = new PointInTimeReference.Builder().id(id);
                    List<FieldValue> finalSort = sort;
                    if (CollectionUtils.isNotEmpty(finalSort)) {
                        pidSearch = elasticsearchClient.search(s -> s
                                        .pit(pBuilder.build()).query(query).trackScores(false).source(sc -> sc.fetch(false)).sort(so ->
                                                so.field(f -> f.field("_doc").order(SortOrder.Asc))).searchAfter(finalSort).trackScores(false).size(10000),
                                HashMap.class
                        );
                    } else {
                        pidSearch = elasticsearchClient.search(s -> s
                                        .pit(pBuilder.build()).query(query).trackScores(false).source(sc-> sc.fetch(false)).sort(so ->
                                                so.field(f -> f.field("_doc").order(SortOrder.Asc))).trackScores(false).size(10000),
                                HashMap.class
                        );
                    }


                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ElasticsearchException e) {
                    log.error("query one start--->", e);
                }
                if (pidSearch.hits().hits().size() == 0) {
                    return;
                }
                n++;
                nLong.addAndGet(1);
                //processSearchHitsListHit(scrollSearch.hits().hits(), bitSet);

                sort = getSortAfter(pidSearch.hits().hits(), bitSet);
                timeLong.addAndGet(pidSearch.hits().hits().size());

                log.info(" {}----start----{}---->cost#############{}--query-one--time-avg--{}", Thread.currentThread().getName(),
                        n, (st - System.currentTimeMillis()), (startTime - System.currentTimeMillis()) / nLong.get());
                log.info("avg=========>total------->{}->,avg------>{}/s ", timeLong.get(), (timeLong.get() / ((System.currentTimeMillis() + 1000 - startTime) / 1000)));

            } while (!pidSearch.hits().hits().isEmpty());

            // 清除滚动上下文
            String finalId = id;
            elasticsearchClient.closePointInTime(builder -> builder.id(finalId));

        } catch (Exception e) {
            e.printStackTrace();
        }
        stopWatch.stop();
        log.info("total time count(s):---->" + stopWatch.getTotalTimeSeconds());
    }


    private void queryOnce(String indexName, Query query, int len) throws IOException {


        CountResponse count = elasticsearchClient.count(s -> s.index(indexName));

        log.info("start-total-num----" + count.count());
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            OpenPointInTimeRequest.Builder openBuilder = new OpenPointInTimeRequest.Builder();
            openBuilder.index(IndexName.USER_LABEL);
            openBuilder.keepAlive(t -> t.time("10m"));
            String id = elasticsearchClient.openPointInTime(openBuilder.build()).id();
            try {

                PointInTimeReference.Builder pBuilder = new PointInTimeReference.Builder().id(id);
                SearchResponse<HashMap> pidSearch = elasticsearchClient.search(s -> s
                                .pit(pBuilder.build()).query(query).trackScores(false).source(sc -> sc.fetch(false)).sort(so ->
                                        so.field(f -> f.field("_doc").order(SortOrder.Asc))).trackScores(false).size(10000),
                        HashMap.class
                );

                stopWatch.stop();
                log.info("pit-condition--{} must 9 should----cost:---->{}----->size---{}",len, stopWatch.getTotalTimeSeconds(), pidSearch.hits().hits().size());

            } catch (IOException e) {
                throw new RuntimeException(e);
            } catch (ElasticsearchException e) {
                log.error("query one start--->", e);
            }
            elasticsearchClient.closePointInTime(builder -> builder.id(id));

            stopWatch.start();
            SearchResponse<HashMap> fromSearch = elasticsearchClient.search(s -> s
                            .query(query).trackScores(false).source(sc -> sc.fetch(false)).sort(so ->
                                    so.field(f -> f.field("_doc").order(SortOrder.Asc))).trackScores(false).size(10000),
                    HashMap.class
            );
            stopWatch.stop();
            log.info("from--condition--{} must 9 should-cost:---->{}----->size---{}", len, stopWatch.getTotalTimeSeconds(), fromSearch.hits().hits().size());

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private List<FieldValue> getSortAfter(List<Hit<HashMap>> hits, BitSet bitSet) {
        // 处理查询结果
        Hit<HashMap> hashMapHit = hits.get(hits.size() - 1);
        log.info(hashMapHit.id());
        hits.stream().forEach(item -> catCount(bitSet, item));
        List<FieldValue> sort = hashMapHit.sort();
        return sort;
    }

    private void catCount(BitSet bitSet, Hit<HashMap> hit) {
        bitSet.set(Integer.valueOf(hit.id()));
        set.add(hit.id());

    }

    private void processSearchHitsListHit(List<Hit<HashMap>> searchHits, BitSet bitSet) {
        // 处理查询结果
        for (Hit<HashMap> searchHit : searchHits) {
            // 处理单个文档
            Object id = searchHit.id();

            bitSet.set(Integer.valueOf(String.valueOf(id)));
        }
    }

    public void complex() {

        Query.Builder queryBuilder = new Query.Builder();
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        // must
        boolQueryBuilder.must(q -> q.term(builder -> builder.field("candidateurgentscore").value("推送次数≤10次")));
        boolQueryBuilder.must(q ->
                q.range(r -> r.field("last_second_last_login_date").gte(JsonData.of("1306457600000"))
                        .lt(JsonData.of("1806457600000"))));

        // should
        boolQueryBuilder.should(q -> q.term(builder ->
                builder.field("c_school_name_last").value("中国工运学院")));

        FieldValue of1 = FieldValue.of("0");
        FieldValue of2 = FieldValue.of("1");
        List<FieldValue> fieldValues = new ArrayList<>();
        fieldValues.add(of1);
        fieldValues.add(of2);
  /*      boolQueryBuilder.should(q -> q.bool(b ->
                b.must(t-> t.term(c->c.field("is_oversea_second_last").value("0")))
                        .should(t-> t.term(c->c.field("is_oversea_second_last").value("1")))));*/
        boolQueryBuilder.should(q -> q.terms(t -> t.field("is_oversea_second_last").terms(a -> a.value(fieldValues))));
        /*BoolQueryBuilder shouldBoolQueryBuilder = QueryBuilders.boolQuery();
        shouldBoolQueryBuilder.must(QueryBuilders.matchQuery("field3", "value3"));
        boolQueryBuilder.should(shouldBoolQueryBuilder);*/


        // filter
        boolQueryBuilder.filter(q -> q.term(builder -> builder.field("prac_cname").value("有上进心")));
/*        boolQueryBuilder.filter(QueryBuilders.boolQuery()
                .mustNot(QueryBuilders.existsQuery("field4")));*/
        boolQueryBuilder.should(t -> t.wildcard(w -> w.field("bonus_cname").wildcard("*本人*")));
        BitSet bitSet = new BitSet();
        BoolQuery build = boolQueryBuilder.build();

        queryBuilder.bool(build);
        System.out.println("boolStr------------------->" + build);
        complexDo(bitSet, queryBuilder.build());

    }

    public void complexDo(BitSet bitSet, Query query) {
        log.info("开始滚动查询----");
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {

            SearchResponse<HashMap> searchResponse = elasticsearchClient.search(s -> s
                            .index(IndexName.USER_LABEL)
                            .query(query).trackScores(false).source(sc -> sc.fetch(false)).size(10000).scroll(t -> t.time("10s")),
                    HashMap.class
            );

            //使用scroll深度分页查询

            // 处理第一次查询结果
            processSearchHitsListHit(searchResponse.hits().hits(), bitSet);
            // 获取滚动 ID
            String scrollId = searchResponse.scrollId();

            ScrollResponse<HashMap> scrollSearch = null;
            do {
                String finalScrollId = scrollId;
                try {
                    scrollSearch = elasticsearchClient.scroll(s -> s.scrollId(finalScrollId).scroll(t -> t.time("100s")), HashMap.class);

                    log.info("query one start----{}", LocalDateTime.now());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ElasticsearchException e) {
                    log.error("query one start--->", e);
                }

                processSearchHitsListHit(scrollSearch.hits().hits(), bitSet);

                scrollId = scrollSearch.scrollId();
            } while (!scrollSearch.hits().hits().isEmpty());

            // 清除滚动上下文
            //清除 scroll
            String finalScrollId1 = scrollId;
            elasticsearchClient.clearScroll(c -> c.scrollId(finalScrollId1));

        } catch (Exception e) {
            e.printStackTrace();
        }
        stopWatch.stop();
        log.info("total time count(s)：" + stopWatch.getTotalTimeSeconds());
    }

    public void complexDoSort(BitSet bitSet, Query query) {
        log.info("开始滚动查询----");
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {

            SearchResponse<HashMap> searchResponse = elasticsearchClient.search(s -> s
                            .index(IndexName.USER_LABEL).query(query)
                            .searchAfter(FieldValue.of(1878133432)).size(10000).scroll(t -> t.time("10s")),
                    HashMap.class
            );

            //使用scroll深度分页查询

            // 处理第一次查询结果
            processSearchHitsListHit(searchResponse.hits().hits(), bitSet);
            // 获取滚动 ID
            String scrollId = searchResponse.scrollId();

            ScrollResponse<HashMap> scrollSearch = null;
            do {
                String finalScrollId = scrollId;
                try {
                    scrollSearch = elasticsearchClient.scroll(s -> s.scrollId(finalScrollId).scroll(t -> t.time("100s")), HashMap.class);

                    log.info("query one start----{}", LocalDateTime.now());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ElasticsearchException e) {
                    log.error("query one start--->", e);
                }

                processSearchHitsListHit(scrollSearch.hits().hits(), bitSet);

                scrollId = scrollSearch.scrollId();
            } while (!scrollSearch.hits().hits().isEmpty());

            // 清除滚动上下文
            //清除 scroll
            String finalScrollId1 = scrollId;
            elasticsearchClient.clearScroll(c -> c.scrollId(finalScrollId1));

        } catch (Exception e) {
            e.printStackTrace();
        }
        stopWatch.stop();
        log.info("total time count(s)：" + stopWatch.getTotalTimeSeconds());
    }

    public void multiThread(int threadNum, int index, int type) {
        List<LabelDto> labelList = new ArrayList<LabelDto>() {{

            add(LabelDto.init("first_post_domain", "51job_wxmp"));
            add(LabelDto.init("b_chatpreference", "1"));
            add(LabelDto.init("yjs_major", "西南科技大学"));
            add(LabelDto.init("candidatepopular", "推送次数≤10次"));
            add(LabelDto.init("yjs_degree", "优秀大学"));
            add(LabelDto.init("last_login_cooperator", "自然流量"));
            add(LabelDto.init("c_school_name_last", "北京工商管理专修学院"));
            add(LabelDto.init("last_login_cooperator", "自然流量"));
            add(LabelDto.init("last_login_domain", "51job_app"));
            add(LabelDto.init("best_resume_num", "1"));
            add(LabelDto.init("b_isblackrsm", "0"));
            add(LabelDto.init("first_post_domain", "2.50"));
            add(LabelDto.init("b_rsmcompetitiveness", "1"));
            add(LabelDto.init("user_redister_partner", "4"));
            add(LabelDto.init("b_bachelorschooltype", "0"));
            add(LabelDto.init("b_last7daysactivechatcount", "0"));
            add(LabelDto.init("resume_key_word", "办公室销售"));
            add(LabelDto.init("company_size_second_last", "2"));
            add(LabelDto.init("sex", "0"));
            add(LabelDto.init("sex", "1"));
            add(LabelDto.init("is_del_user_info", "0"));
            add(LabelDto.init("last_hr_opt_date", "51job_other"));
            add(LabelDto.init("first_bind_enterprise_wechat", "1.59"));
            add(LabelDto.init("expect_func_three", "0829"));

        }};
        if (type == 1) {
            labelList = new ArrayList<LabelDto>() {{
                add(LabelDto.init("test_label_a", "一个月登录过"));
                add(LabelDto.init("test_label_b", "手机用户"));
                add(LabelDto.init("test_label_c", "最近7天登录"));
                add(LabelDto.init("test_label_d", "推送次数≤10次"));
                add(LabelDto.init("test_label_e", "优秀大学"));
                add(LabelDto.init("test_label_f", "自然流量"));

            }};
        }

        long startTime = System.currentTimeMillis();

        if (index > 0) {
            try {
                List<LabelDto> finalLabelList = labelList;
                Query query = Query.of(t -> t.term(v -> v.field(finalLabelList.get(index).getName()).value(finalLabelList.get(index).getValue())));
                pitComplexQuery(IndexName.USER_LABEL, query, new BitSet(), startTime);
                //performScrollQuery(IndexName.USER_LABEL, labelList.get(index), new BitSet(), startTime);
                return;
            } catch (IOException e) {
                log.info("error------------------->", e);
                return;
            }
        }
        ExecutorService customThreadPool = Executors.newFixedThreadPool(threadNum);

        Map<String, List<String>> excelMap = ReadExcelData.getExcelMap();
        labelList.forEach(item -> {
            try {
                CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                    try {
                        performScrollQuery(IndexName.USER_LABEL, item, new BitSet(), startTime);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                    return null;
                }, customThreadPool);

            } catch (Exception e) {
                log.info("error------------------->", e);
                // TODO这里一定捕获异常并睡眠一会
            }
        });
        atomicLong.set(0);

    }

    public void complexMultiThread(int threadNum, int len, int type, int qtype) {
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < threadNum; i++) {
            CompletableFuture.supplyAsync(() -> {
                Query query = builderQuery(len, type, qtype);
                try {
                    pitComplexQuery(IndexName.USER_LABEL, query, new BitSet(), startTime);
                    //complexQuery(IndexName.USER_LABEL, query, new BitSet(), startTime);
                } catch (IOException e) {
                    log.info("error------------------->", e);
                }
                return null;
            }, executorService);
        }

    }

    private Query builderQuery(int len, int type, int qtype) {

        List<LabelDto> labelList = new ArrayList<LabelDto>() {{

            add(LabelDto.init("first_post_domain", "51job_wxmp"));
            add(LabelDto.init("b_chatpreference", "1"));
            add(LabelDto.init("yjs_major", "西南科技大学"));
            add(LabelDto.init("candidatepopular", "推送次数≤10次"));
            add(LabelDto.init("yjs_degree", "优秀大学"));
            add(LabelDto.init("last_login_cooperator", "自然流量"));
            add(LabelDto.init("c_school_name_last", "北京工商管理专修学院"));
            add(LabelDto.init("last_login_cooperator", "自然流量"));
            add(LabelDto.init("last_login_domain", "51job_app"));
            add(LabelDto.init("best_resume_num", "1"));
            add(LabelDto.init("b_isblackrsm", "0"));
            add(LabelDto.init("first_post_domain", "2.50"));
            add(LabelDto.init("b_rsmcompetitiveness", "1"));
            add(LabelDto.init("user_redister_partner", "4"));
            add(LabelDto.init("b_bachelorschooltype", "0"));
            add(LabelDto.init("b_last7daysactivechatcount", "0"));
            add(LabelDto.init("resume_key_word", "办公室销售"));
            add(LabelDto.init("company_size_second_last", "2"));
            add(LabelDto.init("sex", "0"));
            add(LabelDto.init("sex", "1"));
            add(LabelDto.init("is_del_user_info", "0"));
            add(LabelDto.init("last_hr_opt_date", "51job_other"));
            add(LabelDto.init("first_bind_enterprise_wechat", "1.59"));
            add(LabelDto.init("expect_func_three", "0829"));

        }};
        List<LabelDto> labelList3 = new ArrayList<LabelDto>();
        if (type == 1) {
            labelList = new ArrayList<LabelDto>() {{
                add(LabelDto.init("test_label_1", "最近1天登录"));
                add(LabelDto.init("test_label_2", "最近2天登录"));
                add(LabelDto.init("test_label_3", "最近3天登录"));
                add(LabelDto.init("test_label_4", "最近4天登录"));
                add(LabelDto.init("test_label_5", "最近5天登录"));
                add(LabelDto.init("test_label_6", "最近6天登录"));
                add(LabelDto.init("test_label_7", "最近7天登录"));
                add(LabelDto.init("test_label_8", "最近8天登录"));
                add(LabelDto.init("test_label_9", "最近9天登录"));
                add(LabelDto.init("test_label_10", "最近10天登录"));
                add(LabelDto.init("test_label_11", "最近11天登录"));
                add(LabelDto.init("test_label_12", "最近12天登录"));
                add(LabelDto.init("test_label_13", "最近13天登录"));
                add(LabelDto.init("test_label_14", "最近14天登录"));
                add(LabelDto.init("test_label_15", "最近15天登录"));
                add(LabelDto.init("test_label_16", "最近16天登录"));
                add(LabelDto.init("test_label_17", "最近17天登录"));
                add(LabelDto.init("test_label_18", "最近18天登录"));
                add(LabelDto.init("test_label_19", "最近19天登录"));


            }};
            labelList3 = new ArrayList<LabelDto>() {{
                add(LabelDto.init("test_1", "工作1年"));
                add(LabelDto.init("test_2", "工作2年"));
                add(LabelDto.init("test_3", "工作3年"));
                add(LabelDto.init("test_4", "工作4年"));
                add(LabelDto.init("test_5", "工作5年"));
                add(LabelDto.init("test_6", "工作6年"));
                add(LabelDto.init("test_7", "工作7年"));
                add(LabelDto.init("test_8", "工作8年"));
                add(LabelDto.init("test_10", "工作10年"));
            }};
        }


        List<LabelDto> labelList2 = new ArrayList<LabelDto>() {{

            LabelDto testLabelNum = LabelDto.init("test_label_num_d", "1");
            List<Integer> valuesList = new ArrayList<Integer>() {{
                add(-2000000000);
            }};
            testLabelNum.setData(valuesList);
            testLabelNum.setType("gte");
            add(testLabelNum);

        }};

        Query.Builder queryBuilder = new Query.Builder();

       /*
        QueryStringQuery.Builder   queryStringQueryBuilder = new QueryStringQuery.Builder();
        queryStringQueryBuilder.fields(null).query(null);
        QueryStringQuery build1 = queryStringQueryBuilder.build();

        SearchRequest searchRequest = new SearchRequest.Builder();
         elasticsearchClient.search(s -> s.index("").query(build1))
*/
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        if (qtype != 1) {
            boolQueryBuilder.must(getList(labelList, len));
            boolQueryBuilder.should(getList(labelList3, len));
        } else {
            boolQueryBuilder.filter(getList(labelList, len));
        }

        boolQueryBuilder.mustNot(getList(labelList2, 1));
        queryBuilder.bool(boolQueryBuilder.build());
        Query query = queryBuilder.build();
        log.info("query-str---------------------------->{}", query);
        return query;
    }

    private List<Query> getList(List<LabelDto> labelList, int count) {
        List<Query> queries = new ArrayList<>();
        List<LabelDto> randomElements = LabelBuilder.getRandomElements(labelList, count);
        randomElements.forEach(item -> {
            Query query = null;
            if (StringUtils.isEmpty(item.getType()) || "keyword".equals(item.getType())) {
                TermQuery.Builder termQueryBuilder = new TermQuery.Builder();
                termQueryBuilder.field(item.getName()).value(item.getValue());
                TermQuery termQuery = termQueryBuilder.build();
                query = Query.of(builder -> builder.term(termQuery));
            } else if ("gte".equals(item.getType())) {
                List<Integer> data = (List<Integer>) item.getData();
                ;
                query = RangeQuery.of(r -> r
                        .field(item.getName())
                        .gte(JsonData.of(data.get(0)))
                )._toQuery();
            } else {
                /*WildcardQuery.Builder wildcardQueryBuilder = new WildcardQuery.Builder();
                wildcardQueryBuilder.field(item.getName()).value(item.getValue());
                WildcardQuery wildcardQuery = wildcardQueryBuilder.build();
                query = Query.of(builder -> builder.wildcard(wildcardQuery));*/
/*
                query = WildcardQuery.of(r -> r.field(item.getName()).wildcard(item.getValue()))._toQuery();
*/
            /*    FieldValue of1 = FieldValue.of("0");
                FieldValue of2 = FieldValue.of("1");
                List<FieldValue> fieldValues = new ArrayList<>();
                fieldValues.add(of1);
                fieldValues.add(of2);
                query = TermsQuery.of(r -> r.field(item.getName()).terms(t -> t.value(fieldValues)))._toQuery();*/
/*
                query = MatchQuery.of(r -> r.field(item.getName()).query(item.getValue()))._toQuery();
*/
/*
                query = MatchPhraseQuery.of(r -> r.field(item.getName()).query(item.getValue()))._toQuery();
*/


            }
            BoolQuery.Builder boolBuilder = new BoolQuery.Builder();
            BoolQuery bool = query.bool();
            queries.add(bool._toQuery());

            queries.add(query);
        });

        return queries;
    }

    public void qop(int threadNum, int len, int type, int qtype) {
        Query query = builderQuery(len, type, qtype);
        try {
            queryOnce(IndexName.USER_LABEL, query, len);
        } catch (IOException e) {
            log.info("error------------------->", e);
        }
    }

}
